/**
* <p>Title: HbaseTest.java</p>
* <p>Description: </p>
* <p>Company: www.huoyunren.com</p>
* @author duanxiangyun
* @date 2017年3月21日
* @version 1.0
*/
package com.smarttest.common.util;


import java.io.IOException;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;

public class HbaseUtils {
    // 声明静态配置
    private  Configuration conf = null;
    public HbaseUtils(Map<String, String> confMap) {
    	conf =  HBaseConfiguration.create();
		for (Entry<String, String> entry : confMap.entrySet()) {
			conf.set(entry.getKey(), entry.getValue());
		}
    }
    /*
     * 创建表
     * 
     * @tableName 表名
     * 
     * @family 列族列表
     */
    public void creatTable(String tableName, String[] family)
            throws Exception {
        HBaseAdmin admin = new HBaseAdmin(conf);
        HTableDescriptor desc = new HTableDescriptor(tableName);
        for (int i = 0; i < family.length; i++) {
            desc.addFamily(new HColumnDescriptor(family[i]));
        }
        if (admin.tableExists(tableName)) {
            System.out.println("table Exists!");
            System.exit(0);
        } else {
            admin.createTable(desc);
            System.out.println("create table Success!");
        }
    }

    /*
     * 为表添加数据（适合知道有多少列族的固定表）
     * 
     * @rowKey rowKey
     * 
     * @tableName 表名
     * 
     * @column1 第一个列族列表
     * 
     * @value1 第一个列的值的列表
     * 
     * @column2 第二个列族列表
     * 
     * @value2 第二个列的值的列表
     */
    public void addData(String rowKey, String tableName,
            String[] column1, String[] value1, String[] column2, String[] value2)
            throws IOException {
        Put put = new Put(Bytes.toBytes(rowKey));// 设置rowkey
        HTable table = new HTable(conf, Bytes.toBytes(tableName));// HTabel负责跟记录相关的操作如增删改查等//
                                                                    // 获取表
        HColumnDescriptor[] columnFamilies = table.getTableDescriptor() // 获取所有的列族
                .getColumnFamilies();

        for (int i = 0; i < columnFamilies.length; i++) {
            String familyName = columnFamilies[i].getNameAsString(); // 获取列族名
            if (familyName.equals("article")) { // article列族put数据
                for (int j = 0; j < column1.length; j++) {
                    put.add(Bytes.toBytes(familyName),
                            Bytes.toBytes(column1[j]), Bytes.toBytes(value1[j]));
                }
            }
            if (familyName.equals("author")) { // author列族put数据
                for (int j = 0; j < column2.length; j++) {
                    put.add(Bytes.toBytes(familyName),
                            Bytes.toBytes(column2[j]), Bytes.toBytes(value2[j]));
                }
            }
        }
        table.put(put);
        System.out.println("add data Success!");
    }

    /*
     * 根据rwokey查询
     * 
     * @rowKey rowKey
     * 
     * @tableName 表名
     */
    public Result getResult(String tableName, String rowKey)
            throws IOException {
        Get get = new Get(Bytes.toBytes(rowKey));
        HTable table = new HTable(conf, Bytes.toBytes(tableName));// 获取表
        Result result = table.get(get);
        if (!result.isEmpty()) {
        	for (KeyValue kv : result.list()) {
                System.out.println("family:" + Bytes.toString(kv.getFamily()));
                System.out
                        .println("qualifier:" + Bytes.toString(kv.getQualifier()));
                System.out.println("value:" + Bytes.toString(kv.getValue()));
                System.out.println("Timestamp:" + kv.getTimestamp());
                System.out.println("-------------------------------------------");
            }
        }  
        return result;
    }
//
//    public Result getResultEMS(String tableName, String imei, long timestamp) throws IOException {
//    	String rowKey = getReverseString(imei)+timestamp;
//         return getResult(tableName, rowKey);
//    }
//    public ResultScanner getResultEMS(String tableName, String imei, String from,String to) throws IOException, ParseException {
//    	String imeireverse = getReverseString(imei);
//    	SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
//    	String start_rowkey = imeireverse+sdf.parse(from).getTime();
//    	String stop_rowkey = imeireverse+sdf.parse(from).getTime();
//         return getResultScann(tableName, start_rowkey, stop_rowkey);
//    }
    /*
     * 遍历查询hbase表
     * 
     * @tableName 表名
     */
    public List<Map<String, String>> getResultScann(String tableName) throws IOException {
    	System.out.println("duanxiangyunduanxiangyunduanxiangyunduanxiangyunduanxiangyun");
        Scan scan = new Scan();
        ResultScanner rs = null;
        HTable table = new HTable(conf, Bytes.toBytes(tableName));
        List<Map<String,String>> l = new ArrayList<Map<String,String>>();
        try {
            rs = table.getScanner(scan);
            for (Result r : rs) {
            	for (Cell kv : r.listCells()) {
                	Map<String,String> m = new HashMap<String, String>();
                	m.put("rowkey", Bytes.toString(kv.getRow()));
                	m.put("timestamp", String.valueOf(kv.getTimestamp()));
                	m.put("family", Bytes.toString(kv.getFamily()));
                	m.put("qualifier", Bytes.toString(kv.getQualifier()));
                	m.put("value", Bytes.toString(kv.getValue()));
                	l.add(m);
                }
            }
        } finally {
        	if (null != rs) {
                rs.close();
        	}
        }
        return l;
    }

    /*
     * 遍历查询hbase表
     * 
     * @tableName 表名
     */
    public List<Map<String, String>> getResultScann(String tableName, String start_rowkey,
            String stop_rowkey) throws IOException {
        Scan scan = new Scan();
        scan.setStartRow(Bytes.toBytes(start_rowkey));
        scan.setStopRow(Bytes.toBytes(stop_rowkey));
        ResultScanner rs = null;
        List<Map<String,String>> l = new ArrayList<Map<String,String>>();
        HTable table = new HTable(conf, Bytes.toBytes(tableName));
        try {
            rs = table.getScanner(scan);
            for (Result r : rs) {
                for (Cell kv : r.listCells()) {
                	Map<String,String> m = new HashMap<String, String>();
                	m.put("rowkey", Bytes.toString(kv.getRow()));
                	m.put("timestamp", String.valueOf(kv.getTimestamp()));
                	m.put("family", Bytes.toString(kv.getFamily()));
                	m.put("qualifier", Bytes.toString(kv.getQualifier()));
                	m.put("value", Bytes.toString(kv.getValue()));
                	l.add(m);
                }
            }
        } finally {
        	if (null != rs) {
                rs.close();
        	}
        }
        return l;
    }
 // 获取所有数据
 	public void getAllRows(String tableName) throws Exception {
 		HTable table = new HTable(conf, tableName);
 		Scan scan = new Scan();
 		ResultScanner results = table.getScanner(scan);
 		// 输出结果
 		for (Result result : results) {
 			for (KeyValue rowKV : result.raw()) {
 				System.out.print("key:" + rowKV.getKeyString() + " ");
 				System.out.print("行名:" + new String(rowKV.getRow()) + " ");
 				System.out.print("时间戳:" + rowKV.getTimestamp() + " ");
 				System.out.print("列族名:" + new String(rowKV.getFamily()) + " ");
 				System.out
 						.print("列名:" + new String(rowKV.getQualifier()) + " ");
 				System.out.println("值:" + new String(rowKV.getValue()));
 			}
 		}
 		results.close();
 	}
    /*
     * 查询表中的某一列
     * 
     * @tableName 表名
     * 
     * @rowKey rowKey
     */
    public void getResultByColumn(String tableName, String rowKey,
            String familyName, String columnName) throws IOException {
        HTable table = new HTable(conf, Bytes.toBytes(tableName));
        Get get = new Get(Bytes.toBytes(rowKey));
        get.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(columnName)); // 获取指定列族和列修饰符对应的列
        Result result = table.get(get);
        for (KeyValue kv : result.list()) {
            System.out.println("family:" + Bytes.toString(kv.getFamily()));
            System.out
                    .println("qualifier:" + Bytes.toString(kv.getQualifier()));
            System.out.println("value:" + Bytes.toString(kv.getValue()));
            System.out.println("Timestamp:" + kv.getTimestamp());
            System.out.println("-------------------------------------------");
        }
    }

    /*
     * 更新表中的某一列
     * 
     * @tableName 表名
     * 
     * @rowKey rowKey
     * 
     * @familyName 列族名
     * 
     * @columnName 列名
     * 
     * @value 更新后的值
     */
    public void updateTable(String tableName, String rowKey,
            String familyName, String columnName, String value)
            throws IOException {
        HTable table = new HTable(conf, Bytes.toBytes(tableName));
        Put put = new Put(Bytes.toBytes(rowKey));
        put.add(Bytes.toBytes(familyName), Bytes.toBytes(columnName),
                Bytes.toBytes(value));
        table.put(put);
        System.out.println("update table Success!");
    }

    /*
     * 查询某列数据的多个版本
     * 
     * @tableName 表名
     * 
     * @rowKey rowKey
     * 
     * @familyName 列族名
     * 
     * @columnName 列名
     */
    public void getResultByVersion(String tableName, String rowKey,
            String familyName, String columnName) throws IOException {
        HTable table = new HTable(conf, Bytes.toBytes(tableName));
        Get get = new Get(Bytes.toBytes(rowKey));
        get.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(columnName));
        get.setMaxVersions(5);
        Result result = table.get(get);
        for (KeyValue kv : result.list()) {
            System.out.println("family:" + Bytes.toString(kv.getFamily()));
            System.out
                    .println("qualifier:" + Bytes.toString(kv.getQualifier()));
            System.out.println("value:" + Bytes.toString(kv.getValue()));
            System.out.println("Timestamp:" + kv.getTimestamp());
            System.out.println("-------------------------------------------");
        }
        /*
         * List<?> results = table.get(get).list(); Iterator<?> it =
         * results.iterator(); while (it.hasNext()) {
         * System.out.println(it.next().toString()); }
         */
    }

    /*
     * 删除指定的列
     * 
     * @tableName 表名
     * 
     * @rowKey rowKey
     * 
     * @familyName 列族名
     * 
     * @columnName 列名
     */
    public void deleteColumn(String tableName, String rowKey,
            String falilyName, String columnName) throws IOException {
        HTable table = new HTable(conf, Bytes.toBytes(tableName));
        Delete deleteColumn = new Delete(Bytes.toBytes(rowKey));
        deleteColumn.deleteColumns(Bytes.toBytes(falilyName),
                Bytes.toBytes(columnName));
        table.delete(deleteColumn);
        System.out.println(falilyName + ":" + columnName + "is deleted!");
    }

    /*
     * 删除指定的列
     * 
     * @tableName 表名
     * 
     * @rowKey rowKey
     */
    public void deleteAllColumn(String tableName, String rowKey)
            throws IOException {
        HTable table = new HTable(conf, Bytes.toBytes(tableName));
        Delete deleteAll = new Delete(Bytes.toBytes(rowKey));
        table.delete(deleteAll);
        System.out.println("all columns are deleted!");
    }

    /*
     * 删除表
     * 
     * @tableName 表名
     */
    public void deleteTable(String tableName) throws IOException {
        HBaseAdmin admin = new HBaseAdmin(conf);
        admin.disableTable(tableName);
        admin.deleteTable(tableName);
        System.out.println(tableName + "is deleted!");
    }
    /**
     * 获得倒序字符串
     * @param str
     * @return
     */
    public static String getReverseString(String str) {
    	return new StringBuffer(str).reverse().toString();
    }
    /**
     * 给字符串补位
     * @param str
     * @return
     */
    public static String autoCover(String str, int len, int loc, String mark) {
    	String returnstr = "";
    	if (str.length() < len) {
    		String markstr = "";
			for (int i = 0; i < len - str.length(); i++) {
				markstr += mark;
			}
    		if (loc == 0) {
    			returnstr = markstr+str;
    		}
    		else {
    			returnstr = str + markstr;
    		}
    	}
    	return returnstr;
    }
    
    /**
    * MD5 32位加密
    * @param text 明文
    * @return 32位MD5密文
    */
   public static String MD5(String text) {
       String re_md5 = new String();
       try {
           MessageDigest md = MessageDigest.getInstance("MD5");
           md.update(text.getBytes());
           byte b[] = md.digest();

           int i;

           StringBuffer buf = new StringBuffer("");
           for (int offset = 0; offset < b.length; offset++) {
               i = b[offset];
               if (i < 0)
                   i += 256;
               if (i < 16)
                   buf.append("0");
               buf.append(Integer.toHexString(i));
           }

           re_md5 = buf.toString();

       } catch (NoSuchAlgorithmException e) {
           e.printStackTrace();
       }
       return re_md5;
   }
}
